package com.simple.datax.seda.internal;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.simple.datax.seda.api.ConfigInfoIF;
import com.simple.datax.seda.api.DealSourceIF;
import com.simple.datax.seda.core.exception.InitFailedException;
import com.simple.datax.seda.util.StringUtil;

public class LinkedDealSource implements DealSourceIF {

	private static final Log LOG = LogFactory.getLog(LinkedDealSource.class);

	private String dealIndex = "";

	private ConfigInfoIF configInfo;

	private final Object LOCKER = new Object();

	private ArrayList<DealSourceIF> dealers = new ArrayList<DealSourceIF>();

	public LinkedDealSource() {
	}

	public void initialize() throws InitFailedException {
		String[] dealerNames = StringUtil.split(this.dealIndex, ";");
		if (dealerNames == null || dealerNames.length < 1) {
			throw new InitFailedException("数据处理器列表是空,请检查dealIndex属性的设置是否正确");
		}
		this.clear();
		synchronized (LOCKER) {
			try {
				int count = 0;
				for (int i = 0; i < dealerNames.length; i++) {
					if (dealerNames[i].trim().length() == 0) {
						continue;
					}

					DealSourceIF dealer = (DealSourceIF) this.configInfo
							.getObject(dealerNames[i]);
					if (dealer == null) {
						throw new InitFailedException("创建处理器[" + dealerNames[i]
								+ "失败.");
					}
					dealer.setCustomConfigurator(configInfo);
					dealer.initialize();
					dealers.add(dealer);
					count++;
				}
				if (count == 0) {
					LOG.warn("没有配置数据的后续处理器,请管理员确定是有意这样配置还是因为疏忽忘记配置.");
				}
			} catch (Exception e) {
				throw new InitFailedException("创建用户可定制数据处理器失败:"
						+ e.getMessage());
			}
		}

	}

	public void setCustomConfigurator(ConfigInfoIF config) {
		configInfo = config;
	}

	public void clear() {
		synchronized (LOCKER) {
			if (dealers != null) {
				Iterator<DealSourceIF> iter = dealers.iterator();
				while (iter.hasNext()) {
					iter.next().clear();
				}
				dealers.clear();
			}
		}

	}

	public Object deal(Object dealObj) throws Exception {
		synchronized (LOCKER) {
			if (dealers.size() == 0)
				return dealObj;
			Object item = dealObj;
			for (int i = 0; i < dealers.size(); i++) {
				item = dealers.get(i).deal(item);
			}
			return item;
		}
	}

	public String getDealIndex() {
		return dealIndex;
	}

	public void setDealIndex(String dealIndex) {
		this.dealIndex = dealIndex;
	}

}
